package org.ws.task.item.support;

import org.apache.commons.collections4.CollectionUtils;
import org.ws.task.item.AbstractTask;
import org.ws.task.item.PushTask;
import org.ws.task.listener.item.support.DefaultItemCommitListener;
import org.ws.task.model.protocol.PushItemRequest;
import org.ws.task.model.protocol.PushItemResponse;
import org.ws.task.model.TaskItem;

import java.util.Collection;
import java.util.function.Consumer;

/**
 * 推数据:
 * 任一节点将数据推送到服务器端统一调度分配
 * @param <T>
 */
public class SimplePushTask<T> extends AbstractTask implements PushTask<T> {

    public SimplePushTask(String name) {
        super(name);
        this.registerListener(new DefaultItemCommitListener());
    }

    @Override
    public PushItemResponse pushSync(Collection<TaskItem<T>> items, long timeout) {
        if(CollectionUtils.isEmpty(items)){
            PushItemResponse pushItemResponse = new PushItemResponse();
            pushItemResponse.setSuccess(false);
            pushItemResponse.setMessage("empty items.");
            return pushItemResponse;
        }
        this.awaitStart(30_000);
        PushItemRequest request = new PushItemRequest();
        request.setItems(items);
        request.setTaskName(name());

        return this.remoteAccessApi.pushItemSync(request, timeout);
    }

    @Override
    public void pushAsync(Collection<TaskItem<T>> items, long timeout, Consumer<PushItemResponse> callback) {
        if(CollectionUtils.isEmpty(items)){
            return;
        }
        this.awaitStart(30_000);
        PushItemRequest request = new PushItemRequest();
        request.setItems(items);
        request.setTaskName(name());
        this.remoteAccessApi.pushItemAsync(request,timeout,callback);
    }
}
